package utils

import (
	"errors"
	"fmt"
	amqp "github.com/rabbitmq/amqp091-go"
	"sync"
)

type AmqpClient struct {
	url         string
	Connection  *amqp.Connection
	channelPool sync.Pool
	once        sync.Once
	lock        sync.RWMutex
}

func NewAmqpClient(url string) (*AmqpClient, error) {
	conn, err := amqp.Dial(url)
	if err != nil {
		return nil, err
	}
	return &AmqpClient{
		url:        url,
		Connection: conn,
	}, nil
}

func (ac *AmqpClient) Close() error {
	if ac.Connection.IsClosed() {
		return nil
	}
	return ac.Connection.Close()
}

func (ac *AmqpClient) GetChannel() (*amqp.Channel, error) {
	ac.once.Do(func() {
		ac.channelPool = sync.Pool{New: func() any {
			var ch *amqp.Channel
			var err error
			if ac.Connection.IsClosed() {
				ac.lock.Lock()
				ac.Connection, err = amqp.Dial(ac.url)
				ac.lock.Unlock()
				if err != nil {
					return err
				}
			}
			ac.lock.RLock()
			defer ac.lock.RUnlock()
			i := 0
		getChannel:
			if ch, err = ac.Connection.Channel(); err != nil {
				i++
				if !errors.Is(err, amqp.ErrClosed) || i > 3 {
					return err
				}
				goto getChannel
			}
			return ch
		}}
	})
tryAgain:
	ich := ac.channelPool.Get()
	if ch, ok := ich.(*amqp.Channel); ok {
		if ch.IsClosed() {
			goto tryAgain
		}
		return ch, nil
	} else if err, ok := ich.(error); ok {
		return nil, err
	}
	return nil, fmt.Errorf("fail to get channel")
}

func (ac *AmqpClient) PutChannel(ch *amqp.Channel) {
	if ch == nil || ch.IsClosed() {
		return
	}
	ac.channelPool.Put(ch)
}
